package com.nulldev.util.internal.backport.httpclient_rw.httpclient;

import java.util.Map;

import com.nulldev.util.JVM.Arguments;
import com.nulldev.util.data.Variables;
import com.nulldev.util.data.Arrays.maps.FastMaps;
import com.nulldev.util.logging.Logger;
import com.nulldev.util.logging.LoggerFactory;

public class BackportInputStreamMonitor extends Thread implements Runnable {

	private static BackportInputStreamMonitor INSTANCE;

	private static final Logger log = LoggerFactory.getLoggerDT(Arguments.hasArgument("--debug", "--trace", "--HttpClient.debug"));
	/* Allow streams to stay near-dead for ten seconds. */
	private static final long THRESHOLD = 10 * 1000;

	public static BackportInputStreamMonitor get() {
		if (INSTANCE == null) {
			INSTANCE = new BackportInputStreamMonitor();
		}

		return INSTANCE;
	}

	private final Map<Integer, BackportInputStreamMonitored> ENTRIES = FastMaps.map();

	BackportInputStreamMonitor() {
		super("BACKPORT-IS-MON");
		super.start();
	}

	public void register(final BackportInputStreamMonitored entry) {
		Variables.requireNonNull(entry, "entry");
		this.ENTRIES.put(entry.hashCode(), entry);
		interrupt();
	}

	public void unregister(final BackportInputStreamMonitored entry) {
		Variables.requireNonNull(entry, "entry");

		final int hashCode = entry.hashCode();
		if (!this.ENTRIES.remove(hashCode, entry))
			log.warn("[unregister] Failed to remove an entry... [hash=" + hashCode + ", entry=" + entry + "]");
	}

	@Override
	public void run() {
		while (true) {
			if (ENTRIES.isEmpty()) {
				try {
					/* Sleep for 10 minutes when idle. */
					Thread.sleep(10 * 60 * 1000);
				} catch (InterruptedException e) {
					/* NO-OP */
				}
				continue;
			} else {

				final long test_start = System.currentTimeMillis();
				ENTRIES.forEach((_ignored, stream) -> {
					if (is_stream_stale(test_start, stream)) {
						log.dwarn("Found a stale connection, terminating stream... [diff=" + (test_start - stream.get_last_read_to()) + "ms]");
						stream.killConnection();
					}
				});

				try {
					/* Sleep every THRESHOLD (ten seconds) */
					Thread.sleep(THRESHOLD);
				} catch (InterruptedException e) {
					/* NO-OP */
				}
			}
		}
	}

	private boolean is_stream_stale(final long test_start, final BackportInputStreamMonitored stream) {
		/* This shouldn't ever occur. */
		if (stream == null)
			return false;

		final long last_read = stream.get_last_read_to();
		/* Stream hasn't even been used yet. */
		if (last_read == -1)
			return false;

		return Math.abs(test_start - last_read) >= THRESHOLD;
	}
}
